package com.gtp.mq.demo.spring.config;

import java.util.Iterator;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.aopalliance.aop.Advice;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

/**
 * 负责RabbitTemplate相关管理
 * 不建议自己new对象，这样后期改变配置比较艰难
 *
 * @author gaotingping@janlc.com
 * @date 2019年4月1日
 */
@Component
public class RabbitFactoryConfig implements ApplicationContextAware{
	
	private final static Logger logger=LoggerFactory.getLogger(RabbitFactoryConfig.class);

	@Autowired
	private RabbitTemplate rt;
	
	/**
	 * 统一设置RabbitTemplate
	 * @param rt
	 */
	@PostConstruct
	public void myInit() {
		
		//container.setAfterReceivePostProcessors(null);
		//container.setMessageConverter(null);
		
		/**
		 * 建议开发环境开启,以便于排查消息是否发送成功 ConfirmCallback
		 * 接口用于实现消息发送到RabbitMQ交换器后接收ack回调
		 * 可以回答我消息是否成功发送到服务端了
		 */
		rt.setConfirmCallback(new ConfirmCallback() {
			@Override
			public void confirm(CorrelationData correlationData, boolean ack, String cause) {
				// 发送状态
				logger.info("消费发送:ack={},cause={}", ack, cause);
			}
		});

		/**
		 * ReturnCallback接口用于实现消息发送到RabbitMQ交换器,但无相应队列与交换器绑定时的回调
		 * 可以回答:消息在服务端是否成功匹配到队列了，还是没有找到匹配，丢弃了
		 */
		rt.setReturnCallback(new ReturnCallback() {
			@Override
			public void returnedMessage(Message message, int replyCode, String replyText, String exchange,
					String routingKey) {
				// 无绑定
				logger.info("消费匹配失败:{},{},{}", exchange, routingKey, new String(message.getBody()));
			}
		});

		/**
		 * mandatory 默认为FALSE
		 * 如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，会直接将消息扔掉
		 */
		rt.setMandatory(true);
		
		//可以自定义序列化和反序列化类
		//rt.setMessageConverter(messageConverter);
		
		//可以设置前置消息处理器
		rt.addBeforePublishPostProcessors(new MessagePostProcessor(){

			@Override
			public Message postProcessMessage(Message message) throws AmqpException {
				//消息前置处理器,可以统一处理一些属性设置等
				System.out.println("消息前置处理器-执行(发消息)");
				return message;
			}
			
		});
		
		rt.addAfterReceivePostProcessors(new MessagePostProcessor(){
			@Override
			public Message postProcessMessage(Message message) throws AmqpException {
				//消息前置处理器,可以统一处理一些属性设置等
				System.out.println("消息后置处理器-执行(收消息)");
				return message;
			}
		});
		
		//默认:new SimpleMessageConverter()
		//rt.setMessageConverter(messageConverter);
		
		logger.info("自定义修饰[RabbitTemplate]完成!");
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		
		//按类型获取监听容器
		Map<String, RabbitListenerContainerFactory> map = applicationContext.getBeansOfType(RabbitListenerContainerFactory.class);
		if(map!=null){	
			Iterator<String> it = map.keySet().iterator();
			while(it.hasNext()){
				String k = it.next();
				RabbitListenerContainerFactory container = map.get(k);
				
				logger.info("监听容器:{}={}",k,container);
				
				//direct simple都兼容
				if(container instanceof AbstractRabbitListenerContainerFactory){
					
					AbstractRabbitListenerContainerFactory rc = (AbstractRabbitListenerContainerFactory)container;
				
					//方式1:但是潜在的问题是,容易把别人的覆盖掉
					//rc.setAdviceChain(adviceChain);
					//重试就是通过这个完成的!
					Advice[] ads = rc.getAdviceChain();
					Advice[] newAds=new Advice[ads.length+1];
					
					for(int i=0;i<ads.length;i++){
						newAds[i]=ads[i];
					}
					
					newAds[newAds.length-1]=new MethodInterceptor(){
						@Override
						public Object invoke(MethodInvocation invocation) throws Throwable {
							System.out.println("我是容器级别的Advice!");
							return invocation.proceed();
						}
					};
					
					rc.setAdviceChain(newAds);
					
					//只能设置,不能添加,会不会覆盖原有的,有风险?
					rc.setAfterReceivePostProcessors(new MessagePostProcessor(){

						@Override
						public Message postProcessMessage(Message message) throws AmqpException {
							System.out.println("addAfterReceivePostProcessors-收消息!");
							return message;
						}
						
					});
					
					//pub-sub中都可以定制序列化
					//rc.setMessageConverter(messageConverter);
				}
			}
		}
	}
}
